---
title: HITL 配置指南
description: 配置人机协作工作流的完整指南
---

import { Tabs, TabItem } from '@astrojs/starlight/components';

本指南涵盖所有 HITL 配置选项和模式，用于构建强大的审批工作流。

## HITLConfig 概述

`HITLConfig` 类是配置 HITL 行为的主要接口：

```python
from langcrew.hitl import HITLConfig

config = HITLConfig(
    # 工具级中断
    interrupt_before_tools=["tool1", "tool2"],  # 需要预批准的工具
    interrupt_after_tools=["tool3"],            # 需要后审查的工具
    
    # 任务级中断（仅任务模式）
    interrupt_before_tasks=["task1"],
    interrupt_after_tasks=["task2"],
    
    # 智能体级中断（仅智能体模式）
    interrupt_before_agents=["agent1"],
    interrupt_after_agents=["agent2"],
    
    # 节点级中断（LangGraph 原生）
    interrupt_before_nodes=["node1"],
    interrupt_after_nodes=["node2"]
)
```

## 配置模式

### 指定工具

针对特定工具进行中断：

<Tabs>
<TabItem label="执行前">

```python
# 执行前批准参数
hitl_config = HITLConfig(
    interrupt_before_tools=[
        "web_search",      # 审查搜索查询
        "file_write",      # 批准文件操作
        "send_email",      # 确认邮件发送
        "database_query",  # 验证数据库访问
        "external_api"     # 审查 API 调用
    ]
)
```

</TabItem>
<TabItem label="执行后">

```python
# 执行后审查结果
hitl_config = HITLConfig(
    interrupt_after_tools=[
        "data_analysis",    # 审查分析结果
        "content_generator", # 检查生成内容
        "report_writer",    # 验证报告
        "image_processor"   # 审查处理的图像
    ]
)

# ⚠️ 重要：后置中断仅在单个执行会话内有效
# 工作流重启或从检查点恢复后将不会触发
```

</TabItem>
</Tabs>

## 会话级别限制

### 理解 `interrupt_after_tools` 行为

`interrupt_after_tools` 有一个影响生产部署的重要限制：

```python
# 此配置仅在单个执行会话内有效
hitl_config = HITLConfig(
    interrupt_after_tools=["data_analysis", "report_generator"]
)

# 场景 1：单会话执行 ✅
# 1. 工具执行 → 2. 用户审查结果 → 3. 工作流继续

# 场景 2：工作流重启 ❌  
# 1. 工具执行 → 2. interrupt_after 触发 → 3. 系统重启 → 4. 工具再次执行（重复）
```

**发生原因：**

- `interrupt_after_tools` 发生在工具节点执行过程中，在下一个检查点保存之前
- 系统重启时，无法从中断点恢复，必须从最后一个检查点重新开始
- 工具节点会重新从头开始执行，导致工具再次运行
- 这可能导致工具重复执行和潜在的副作用

### 跨重启审查的解决方案

<Tabs>
<TabItem label="使用前置中断">

```python
# 对关键操作更可靠
hitl_config = HITLConfig(
    interrupt_before_tools=["critical_operation"],  # 总是触发
    # interrupt_after_tools=["critical_operation"]  # 重启后可能被跳过
)
```

</TabItem>
<TabItem label="可重入工具">

```python
from langchain_core.tools import tool

@tool
def reentrant_analysis_tool(data: str, review_count: int = 0) -> str:
    """支持多次审查周期的分析工具"""
    if review_count == 0:
        # 首次执行 - 进行分析
        result = perform_analysis(data)
        return f"分析完成 (审查 #{review_count + 1}): {result}"
    else:
        # 重启后重新执行 - 允许重新审查
        return f"重新审查分析 (审查 #{review_count + 1}): {cached_result}"

# 配置前置中断以确保审查发生
hitl_config = HITLConfig(
    interrupt_before_tools=["reentrant_analysis_tool"]
)
```

</TabItem>
<TabItem label="节点级控制">

```python
# 使用工作流级中断而非工具级
hitl_config = HITLConfig(
    interrupt_after_nodes=["analysis_node"],  # 跨重启有效
    # interrupt_after_tools=["analysis_tool"]  # 会话级限制
)
```

</TabItem>
</Tabs>

## 智能体级配置

### 个人智能体设置

```python
from langcrew import Agent
from langcrew.hitl import HITLConfig

# 高权限智能体，最少中断
admin_agent = Agent(
    role="系统管理员",
    tools=[SystemTool(), FileManagerTool()],
    hitl=HITLConfig(
        interrupt_before_tools=["system_shutdown"]  # 仅关键操作
    )
)

# 普通智能体，标准审批
worker_agent = Agent(
    role="数据工作者",
    tools=[DataProcessorTool(), FileWriteTool()],
    hitl=HITLConfig(
        interrupt_before_tools=["file_write", "data_export"]
    )
)

# 面向公众的智能体，最大监督
public_agent = Agent(
    role="公共助手", 
    tools=[WebSearchTool(), CalculatorTool(), FileWriteTool()],
    hitl=HITLConfig(
        interrupt_before_tools=["web_search", "file_write"]  # 批准所有风险操作
    )
)
```

### 上下文感知配置

```python
def create_context_aware_hitl(user_role, data_sensitivity):
    """根据上下文创建 HITL 配置"""
    
    if user_role == "admin":
        return HITLConfig(
            interrupt_before_tools=["system_operation"]
        )
    elif data_sensitivity == "high":
        return HITLConfig(
            interrupt_before_tools=["file_write", "database_write", "external_api"]
        )
    else:
        return HITLConfig(
            interrupt_before_tools=["file_write", "external_api"]
        )

# 在智能体创建中使用
agent = Agent(
    role="数据处理器",
    hitl=create_context_aware_hitl(
        user_role=current_user.role,
        data_sensitivity=data_classification
    )
)
```

## 团队级配置

### 统一策略

将 HITL 策略应用于多个智能体：

```python
from langcrew import Crew

# 没有 HITL 配置的个人智能体
researcher = Agent(role="研究员", tools=[WebSearchTool()])
writer = Agent(role="作家", tools=[FileWriteTool()])
reviewer = Agent(role="审核员", tools=[EmailTool()])

# 团队级统一策略
crew = Crew(
    agents=[researcher, writer, reviewer],
    hitl=HITLConfig(
        interrupt_before_tools=["web_search", "file_write", "send_email"]
    )
)
```

### 混合配置

结合智能体级和团队级 HITL：

```python
# 有特定 HITL 配置的智能体
critical_agent = Agent(
    role="关键操作",
    tools=[DatabaseTool()],
    hitl=HITLConfig(
        interrupt_before_tools=["database_write", "database_delete"]  # 智能体特定策略
    )
)

# 没有 HITL 配置的智能体（将从团队继承）
regular_agent = Agent(
    role="常规工作者",
    tools=[FileWriteTool()]
)

# 带回退策略的团队
crew = Crew(
    agents=[critical_agent, regular_agent],
    hitl=HITLConfig(
        interrupt_before_tools=["file_write"]  # 仅应用于 regular_agent
    )
)
```

## 节点级中断

与 LangGraph 的原生中断系统集成：

```python
from langcrew.hitl import HITLConfig

# 结合工具和节点中断
hitl_config = HITLConfig(
    # 工具级中断（LangCrew HITL 系统）
    interrupt_before_tools=["critical_operation"],
    interrupt_after_tools=["data_export"],
    
    # 节点级中断（LangGraph 原生）
    interrupt_before_nodes=["decision_node"],
    interrupt_after_nodes=["validation_node"]
)

agent = Agent(
    role="全面智能体",
    tools=[CriticalTool(), DataExportTool()],
    hitl=hitl_config
)
```

## 响应处理

### 简单响应

HITL 接受自然语言响应：

```python
# 这些都表示"批准"：
responses = [
    "yes", "approve", "ok", "confirm", "accept", "agreed",  # 英文
    "批准", "同意", "确认", "通过", "好的", "可以",              # 中文
    True,                                                    # 布尔值
    {"approved": True}                                       # 结构化
]

# 这些都表示"拒绝"：
responses = [
    "no", "deny", "reject", "refuse", "cancel", "disagree", # 英文
    "拒绝", "不同意", "不通过", "取消", "否", "不要",            # 中文
    False,                                                   # 布尔值
    {"approved": False}                                      # 结构化
]
```

### 高级响应

#### 参数修改（前置中断）

```python
# 用户可以修改工具参数
user_response = {
    "approved": True,
    "modified_args": {
        "query": "带有更多上下文的增强搜索查询",
        "max_results": 10,
        "language": "zh",
        "safe_search": True
    }
}
```

#### 结果增强（后置中断）

```python
# 用户可以修改工具结果
user_response = {
    "approved": True,
    "modified_result": """
    原始分析增强内容：
    - 来自领域专业知识的额外上下文
    - 基于最新更新的修正数据点
    - 用户洞察和建议
    """
}
```

#### 带原因的拒绝

```python
# 提供详细的拒绝原因
user_response = {
    "approved": False,
    "reason": "查询参数过于宽泛，可能返回敏感数据"
}
```

## 基于环境的配置

### 开发 vs 生产

```python
import os

def get_hitl_config():
    """获取适合环境的 HITL 配置"""
    
    env = os.getenv("ENVIRONMENT", "development")
    
    if env == "development":
        # 最少中断以加快开发
        return None  # 开发环境无 HITL
    
    elif env == "staging":
        # 仅关键操作
        return HITLConfig(
            interrupt_before_tools=["database_write", "external_api", "file_delete"]
        )
    
    elif env == "production":
        # 全面监督 - 指定所有敏感工具
        return HITLConfig(
            interrupt_before_tools=[
                "database_write", "database_delete", 
                "file_write", "file_delete",
                "external_api", "send_email"
            ]
        )
    
    else:
        # 默认为安全配置
        return HITLConfig(
            interrupt_before_tools=["file_write", "database_query", "external_api"]
        )

# 在智能体创建中使用
agent = Agent(
    role="环境感知智能体",
    hitl=get_hitl_config()
)
```

## 最佳实践

### 1. 安全优先方法

```python
# 从严格开始，指定敏感操作
initial_config = HITLConfig(
    interrupt_before_tools=[
        "file_write", "file_delete",
        "database_write", "database_delete",
        "external_api", "send_email"
    ]
)

# 根据应用需求扩展
production_config = HITLConfig(
    interrupt_before_tools=[
        "file_write", "file_delete",
        "database_write", "database_delete",
        "external_api", "send_email",
        "system_command", "user_data_access"  # 根据需要添加更多
    ]
)
```

### 2. 基于角色的配置

```python
def get_role_based_hitl(user_role):
    """基于用户角色的 HITL 配置"""
    
    role_configs = {
        "admin": HITLConfig(
            interrupt_before_tools=["system_shutdown", "user_delete"]
        ),
        "manager": HITLConfig(
            interrupt_before_tools=["data_export", "report_send", "file_delete"]
        ),
        "analyst": HITLConfig(
            interrupt_before_tools=["database_write", "external_api"]
        ),
        "viewer": HITLConfig(
            interrupt_before_tools=["file_write", "database_write", "external_api"]
        )
    }
    
    return role_configs.get(user_role, role_configs["viewer"])
```

### 3. 工具分类

```python
# 按风险级别分类工具
SAFE_TOOLS = ["calculator", "date_time", "text_formatter", "unit_converter"]
MODERATE_TOOLS = ["web_search", "file_read", "data_query"]  
DANGEROUS_TOOLS = ["file_write", "file_delete", "database_write", "send_email"]

# 低风险：仅危险工具需要批准
low_risk_config = HITLConfig(
    interrupt_before_tools=DANGEROUS_TOOLS
)

# 高风险：危险和中等工具需要批准
high_risk_config = HITLConfig(
    interrupt_before_tools=DANGEROUS_TOOLS + MODERATE_TOOLS
)

# 最大安全性：明确列出所有需要批准的工具
maximum_security_config = HITLConfig(
    interrupt_before_tools=DANGEROUS_TOOLS + MODERATE_TOOLS + ["advanced_operation"]
)
```

## 故障排除

### 常见问题

#### 1. 中断未触发

```python
# ❌ 问题：工具名称不匹配
hitl_config = HITLConfig(interrupt_before_tools=["file_writer"])
agent = Agent(tools=[FileWriteTool()])  # 实际工具名称："file_write"

# ✅ 解决方案：使用确切的工具名称
hitl_config = HITLConfig(interrupt_before_tools=["file_write"])

# 调试：打印工具名称
for tool in agent.tools:
    print(f"工具名称：{tool.name}")
```

#### 2. 后置中断不工作

```python
# ❌ 问题：跨会话使用后置中断
result1 = crew.kickoff(config={"configurable": {"thread_id": "session1"}})  # 工具执行，触发后置中断
# 系统重启...
result2 = crew.kickoff(config={"configurable": {"thread_id": "session1"}})  # 后置中断不会再次触发

# ✅ 解决方案：使用前置中断进行持久审批
hitl_config = HITLConfig(
    interrupt_before_tools=["critical_tool"]  # 总是触发
)
```

#### 3. 响应解析错误

```python
# ❌ 问题：格式错误的响应
user_response = "批准并修改：query=new_query"

# ✅ 解决方案：使用结构化响应
user_response = {
    "approved": True,
    "modified_args": {"query": "new_query"}
}
```

### 调试模式

启用详细日志以进行故障排除：

```python
import logging

# 启用 HITL 调试日志
logging.getLogger("langcrew.hitl").setLevel(logging.DEBUG)

# 创建智能体并运行
agent = Agent(hitl=hitl_config, verbose=True)
result = crew.kickoff(inputs=data)
```

## 下一步

- **[HITL 概念](/zh/concepts/hitl)** - 理解 HITL 架构
- **[快速开始](/zh/guides/hitl/getting-started)** - 快速入门指南
- **示例** - 查看 `examples/components/hitl/` 了解实际实现案例
- **[社区论坛](https://github.com/01-ai/langcrew/discussions)** - 获取帮助和分享经验
